package com.jwater.core.manager.master.check;

import com.jwater.core.manager.common.DataKeys;
import com.jwater.core.manager.common.model.JobInfo;
import com.jwater.core.manager.job.JobScheduler;
import com.jwater.core.manager.master.MasterNodeImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;

/**
 * @description: 检查job是否启动
 * @author: jnner
 * @createDate: 2021/2/5
 * @version: 1.0
 */
public class JobCheck implements Runnable {
    protected static Logger logger = LoggerFactory.getLogger(JobCheck.class);

    private MasterNodeImpl masterNode;
    private Thread thread;

    public JobCheck(MasterNodeImpl masterNode) {
        this.masterNode = masterNode;
    }

    public void start() {
        thread = new Thread(this);
        thread.start();
    }

    @Override
    public void run() {
        while (true) {
            try {
                check();
            } catch (Exception e) {
                logger.error(e.toString(), e);
            }
            synchronized (this) {
                try {
                    this.wait(10000);
                } catch (InterruptedException e) {
                    logger.error(e.toString(), e);
                }
            }
        }
    }

    /**
     * 检查job，只能leader才能调度job执行，非leader需要停止调度器
     */
    public void check() {
        if (masterNode.groupNode == null) {
            return;
        }
        Collection<JobInfo> jobInfos = masterNode.jobConfParse.getJobInfos().values();
        for (JobInfo jobInfo : jobInfos) {
            String jobKey = DataKeys.getJobKey(jobInfo.getRegionName(), jobInfo.getDeployName(), jobInfo.getGroupName(),
                    jobInfo.getJobName());
            String itemKey = DataKeys.getJobGroupKey(jobInfo.getRegionName(), jobInfo.getDeployName(),
                    jobInfo.getGroupName());
            String run = masterNode.getDataItem(DataKeys.KEY_JOB_RUN, jobKey);
            if (run != null) {
                if (masterNode.groupNode.isLeader()) {
                    JobScheduler jobScheduler = masterNode.jobSchedulers.get(itemKey);
                    if (jobScheduler == null) {
                        jobScheduler = new JobScheduler(jobInfo, masterNode.jobConfParse, this.masterNode);
                        masterNode.jobSchedulers.put(itemKey, jobScheduler);
                        jobScheduler.startScheduler();
                    }
                } else {
                    JobScheduler jobScheduler = masterNode.jobSchedulers.get(itemKey);
                    if (jobScheduler != null) {
                        jobScheduler.stopScheduler();
                        masterNode.jobSchedulers.remove(itemKey);
                    }
                }
            }
        }
    }
}
